// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jing

#include "appcoin2/support/feed/fastfeed_util.h"

#include <fmt/format.h>
#include <spdlog/sinks/basic_file_sink.h>
#include <spdlog/spdlog.h>

#include <cmath>
#include <string>
#include <vector>

#include "coin2/app/driver.h"
#include "coin2/base/config/args.h"
#include "coin2/base/file.h"
#include "coin2/base/string_util.h"

// proto
#include "coin2/exchange/feed.pb.h"

uint32_t Digits10(int64_t val)
{
  return static_cast<uint32_t>(std::log10(std::abs(val))) + 1;
}

inline bool ToIntWithGuessPrecions(
    const Channel& channel,
    const std::optional<bool>& precision_ascending,
    double val,
    int64_t suggest_multiplier,
    int64_t& result,
    int64_t& multiplier) {
  // aggregate trades feature accumulate precision error, so 10 * min
  constexpr int64_t max_precision = std::pow(10, std::numeric_limits<double>::digits10);
  constexpr double dbl_min = 10.0 / max_precision;
  multiplier = suggest_multiplier;
  double multiplied = val * multiplier;
  result = std::llround(multiplied);

  bool guess_as_overflow = false;
  bool guess_as_precision = false;
  if (std::fabs(result - multiplied) > (dbl_min * std::pow(10, Digits10(result)))) {
    double divided = multiplied / multiplier;
    while (multiplied > std::numeric_limits<int64_t>::max() ||
           std::fabs(divided - val) > (dbl_min * std::pow(10, Digits10((int64_t)divided)))) {
      // overflow
      guess_as_overflow = true;
      multiplier = multiplier / 10;
      if(multiplier <= 0) {
        LOG_FIRST_N(INFO, 100) << "Multiplier is min " <<
            val << "," << result << "," << multiplier << "," <<
            channel.mea.String() << "." << channel.symbol;
        return true;
      }
      multiplied = val * multiplier;
      divided = multiplied / multiplier;
    }
    if (guess_as_overflow) {
      // recaculate
      result = std::llround(multiplied);
      if (std::fabs(result - multiplied) > (dbl_min * std::pow(10, Digits10(result)))) {
        LOG_FIRST_N(INFO, 100) << "Lost precision as overflow " <<
            val << "," << result << "," << multiplier << "," <<
            channel.mea.String() << "." << channel.symbol;
      }
    } else {
      if (precision_ascending && !precision_ascending.value()) {
        // multiplier aready changed for overflow
        LOG_FIRST_N(INFO, 100) << "Lost precision as overflow " <<
            val << "," << result << "," << multiplier << "," <<
            channel.mea.String() << "." << channel.symbol;
        return false;
      }

      do {
        if (multiplier > 1'000'000'000LL) {
          LOG_FIRST_N(INFO, 100) << "Multiplier is max " <<
              val << "," << result << "," << multiplier << "," <<
              channel.mea.String() << "." << channel.symbol;
          // lost precicison is acceptable
          break;
        }
        guess_as_precision = true;
        multiplier = multiplier * 10;
        multiplied = val * multiplier;
        result = std::llround(multiplied);
      } while (std::fabs(result - multiplied) > (dbl_min * std::pow(10, Digits10(result))));
    }
  }

  return guess_as_overflow || guess_as_precision;
}

inline int64_t ToInt(double val, int64_t multiplier) {
  return std::llround(val * multiplier);
}

bool operator<(const Channel& lhs, const Channel& rhs) {
  return std::tie(lhs.mea, lhs.symbol) < std::tie(rhs.mea, rhs.symbol);
}


void SetupOptions(cxxopts::Options* opt) {
  opt->add_options()("duration", "test duration",
                     cxxopts::value<std::string>());
  opt->add_options()("mea", "MarketExchangeApi", cxxopts::value<std::string>());
  opt->add_options()("recipe", "recipe", cxxopts::value<std::string>());
  opt->add_options()("group", "symbol group", cxxopts::value<std::string>());
  opt->add_options()("norm", "symbol norm", cxxopts::value<std::string>());
  opt->add_options()("worker", "worker number", cxxopts::value<int>());
  opt->add_options()("start", "start time", cxxopts::value<std::string>());
  opt->add_options()("threads", "parse threads number", cxxopts::value<int>());
  opt->add_options()("multiplier", "default double to int multiplier",
                     cxxopts::value<int>()->default_value("1000000000"));
  AddDriverOptions(opt);
  AddProductOptions(opt);
  AddDateOption(opt);
}

void ParseOptionsIntoConfig(const cxxopts::ParseResult& res,
                            DriverConfig* app) {
  ParseDriverOptions(res, app);
  ParseProductOptions(res, app->mutable_feed());
  auto date = ParseDateOption(res);

  // fastfeed-conversion only option
  {
    auto& interval = *app->mutable_feed()
                          ->mutable_common()
                          ->mutable_archive()
                          ->mutable_interval();
    interval.set_start(std::to_string(date));

    if (res.count("start") != 0U) {
      interval.set_start(res["start"].as<std::string>());
    }
    auto start = TimestampFromString(interval.start());

    if (res.count("duration") != 0U) {
      auto duration_str = res["duration"].as<std::string>();
      auto duration = DurationFromString(duration_str);
      interval.set_end(std::to_string(start + duration));
    } else {
      interval.set_end(std::to_string(start + 24 * 60 * 60 * 1'000'000'000LL));
    }
  }

  CHECK_EQ_THROW(res.count("mea"), 1) << "[Error] provide --mea";
  // CHECK_EQ_THROW(res.count("group"), 1) << "[Error] provide --group";
  CHECK_THROW(res.count("group") == 1 || res.count("norm") == 1)
      << "[Error] provide --group or --norm";
  CHECK_EQ_THROW(res.count("recipe"), 1) << "[Error] provide --recipe";
  auto mea_str = res["mea"].as<std::string>();
  std::optional<std::string> group_str_opt, norms_str_opt;
  if (res.count("group") == 1) {
    group_str_opt = res["group"].as<std::string>();
    CHECK_EQ_THROW(group_str_opt.value().size(), 1)
        << "[Error] --group should be of length 1";
  }
  if (res.count("norm") == 1) {
    norms_str_opt = res["norm"].as<std::string>();
  }
  auto recipe = res["recipe"].as<std::string>();
  std::vector<std::string> splitted_recipes;
  splitted_recipes = Split(recipe, ",");
  auto archive_executor = app->mutable_executor()->mutable_archive();
  if (res.count("threads") != 0U) {
    archive_executor->set_threads(res["threads"].as<int>());
  }
  if (archive_executor->threads() > 0) {
    CHECK_EQ_THROW(archive_executor->threads(), splitted_recipes.size())
        << "Threads number must be the same as recipes number!";
  }
  auto exchanges = app->mutable_feed()->mutable_exchanges();
  bool found = false;
  FeedSystemConfig e;
  for (const auto& pair : *exchanges) {
    if (pair.first == mea_str) {
      found = true;
      e = pair.second;
      auto products = e.mutable_products();
      bool query_and_update_symbols = false;
      if (products->has_query_and_update_symbols()){
        query_and_update_symbols = products->query_and_update_symbols();
      }
      products->Clear();
      if (group_str_opt) {
        products->add_groups(group_str_opt.value());
      }
      if (norms_str_opt) {
        std::vector<std::string> splitted_norms =
            Split(norms_str_opt.value(), ",");
        for (const auto& norm_str : splitted_norms) {
          products->add_norms(norm_str);
        }
      }
      if (query_and_update_symbols && mea_str == "Spot.Coingecko.v3"){
        products->set_query_and_update_symbols(query_and_update_symbols);
      }
      e.clear_channels();
      if (splitted_recipes.size() > 1) {
        for (const auto& recipe : splitted_recipes) {
          e.add_archive_recipes(recipe);
        }
      } else {
        e.set_recipe(recipe);
        e.add_archive_recipes(recipe);
      }

      break;
    }
  }
  CHECK_THROW(found) << "[Error] provided mea is not found in config: "
                     << mea_str;
  exchanges->clear();
  (*exchanges)[mea_str] = e;

  // Set worker number
  int worker = 1;
  if (res.count("worker") != 0U) {
    worker = res["worker"].as<int>();
  }
  CHECK_GT_THROW(worker, 0)
      << "[Error] worker number needs to be greater than 0";
  auto& archive = *app->mutable_feed()->mutable_common()->mutable_archive();
  archive.set_worker(worker);

  std::string worker_setting;
  if (archive.arbitration() || archive.shift()) {
    worker_setting = "w0";
  } else {
    worker_setting = "w" + std::to_string(archive.worker());
  }
  // fastfeed conversion log
  {
    auto yymmdd = res["date"].as<std::string>();
    auto& machine = app->feed().common().archive().machine();
    std::string recipes_cat = recipe;
    std::replace(recipes_cat.begin(), recipes_cat.end(), ',', '-');
    auto& start = app->feed().common().archive().interval().start();
    auto group_str = group_str_opt.value_or("_");
    auto path = fmt::format(
        "{}/{}/{}/{}.{}.{}.{}.{}", res["log-dir"].as<std::string>(), yymmdd,
        machine, mea_str, recipes_cat, group_str, worker_setting, start);
    mkdirs_for_file(path);
    auto file_logger = spdlog::basic_logger_mt("file_logger", path);
    spdlog::set_default_logger(file_logger);
    spdlog::flush_on(spdlog::level::warn);
  }
}

void LogAppConfig(const cxxopts::ParseResult& res, const DriverConfig& app) {
  char host[100];
  gethostname(&host[0], 100);
  spdlog::info("mea             : {}", res["mea"].as<std::string>());
  spdlog::info("recipe          : {}", res["recipe"].as<std::string>());
  if (res["group"].count()) {
    spdlog::info("symbol group    : {}", res["group"].as<std::string>());
  }
  if (res["norm"].count()) {
    spdlog::info("symbol norm    : {}", res["norm"].as<std::string>());
  }
  spdlog::info("date            : {}", res["date"].as<std::string>());
  spdlog::info("current host    : {}", host);
}

bool IsBboPriceChanged(
    const FeedUpdate& upd,
    const std::pair<std::optional<double>, std::optional<double>>& last_bbo_price) {
  auto [last_bid_price, last_ask_price] = last_bbo_price;
  
  std::optional<double> bid_price;
  std::optional<double> ask_price;
  std::optional<PriceQty> bid = upd.book().Bid0();
  if (bid) bid_price = bid->price;
  std::optional<PriceQty> ask = upd.book().Ask0();
  if (ask) ask_price = ask->price;
  return last_bid_price != bid_price || last_ask_price != ask_price;
}

FastFeedSubConverter::FastFeedSubConverter(
    const std::string& recipe,
    const std::string& out_dir,
    int multiplier,
    const std::string& pi_dir)
    : recipe_(recipe),
      out_dir_(out_dir),
      default_multiplier_(multiplier),
      pi_dir_(pi_dir),
      start_from_(0),
      publish_bbo_book_only_(false),
      publish_bbo_price_change_(false),
      publish_book_level_(0) {
  is_trade_move2bp_ = (recipe_.find("trade_move2bp") != std::string::npos);
  using ProductHolderManager = coin2::exchange::base::symbology::ProductHolderManager;
  coin::proto::ProductHolderManagerConfig config;
  if (!pi_dir_.empty()) {
    config.set_root_dir(pi_dir_);
  }
  product_holder_manager_ = std::make_unique<ProductHolderManager>(config);
}

void FastFeedSubConverter::Init(::coin2::app::Driver* driver) {
  this->Init(&(driver->config()));
}

void FastFeedSubConverter::Init(const ::coin2::app::DriverConfig* driver_config) {
  // Converter only support single mea execution.
  const auto& config = driver_config->feed();
  int worker = config.common().archive().worker();
  auto it = config.exchanges().begin();
  auto& exchange_config = it->second;
  auto& archive_config = exchange_config.archive();

  int worker_id = 0;
  if (!archive_config.shift() && !archive_config.arbitration()) {
    worker_id = archive_config.worker();
    CHECK_EQ_THROW(worker, worker_id)
        << "fastfeed converter supports only one worker in one run";
  }

  auto mea = MarketExchangeApi::FromString(exchange_config.mea());
  writer_ = std::make_unique<FastFeedWriter>(
      out_dir_,
      archive_config.machine(),
      mea,
      recipe_,
      worker_id,
      ".lz4",
      config.common().archive().interval().start());

  // Warmup period should be handled by parser but ignored by strategy.
  auto interval = FromIntervalConfig(archive_config.interval());
  start_from_ = interval.first;
  proto_ = fastfeed::proto::Feed();
  if (recipe_.find("light_bbo") != std::string::npos) {
    publish_bbo_book_only_ = true;
    if (mea.exchange == ExchangeType::Mexc || mea.exchange == ExchangeType::Bitget){
      publish_book_level_ = 10;
    } else {
      publish_book_level_ = 5;
    }
  } else if (recipe_.find("move2bp") != std::string::npos) {
    publish_bbo_book_only_ = true;
    publish_book_level_ = 5;
  } else if (recipe_ == "bbo_price_change") {
    publish_bbo_price_change_ = true;
    publish_book_level_ = 1;
  } else {
    publish_bbo_book_only_ = exchange_config.raw_parser().publish_bbo_book_only();
    publish_book_level_ = exchange_config.raw_parser().book_level();
  }
}

// When each product's trade or book event appears for the first time
// we write BookBuilderParameter to announce the precision
void FastFeedSubConverter::onNewProduct(
    FastFeedWriter* writer,
    const std::string& symbol,
    int64_t ts,
    int price_precision,
    int qty_precision,
    int default_precision) {
  // write trade/book multipliers
  fastfeed::proto::Feed proto;
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_METADATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_BUILDER_PARAMETER);
  proto.set_symbol(symbol);
  proto.set_timestamp(ts);

  BookBuilderParameter& param = *proto.mutable_book_builder_param();
  param.set_price_precision(price_precision);
  param.set_qty_precision(qty_precision);

  writer->Write(&proto);
}

FastFeedBuilder& FastFeedSubConverter::GetFastFeedBuilder(
    const Channel& ch,
    FastFeedWriter* writer,
    const FeedUpdate& upd) {
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);
  auto p = builders_.insert(std::make_pair(ch, FastFeedBuilder(-SNAPSHOT_PERIOD)));
  if (p.second) {
    onNewProduct(
        writer, ch.symbol, upd.timestamp(),
        precision_multiplier.price_precision,
        precision_multiplier.qty_precision,
        default_multiplier_);
  }
  return p.first->second;
}

const PrecisionMultiplier& FastFeedSubConverter::GetPrecisionMultiplier(
    const Channel& ch,
    const FeedUpdate& upd) {
  if (precision_multiplier_cache_.find(ch) != std::end(precision_multiplier_cache_)) {
    return precision_multiplier_cache_[ch];
  } else {
    try {
      auto product_holder = product_holder_manager_->GetProductHolder(upd.product(), upd.timestamp());
      auto* pi = &(product_holder->product_info());
      int price_precision = pi->tick_price().max_precision();
      int qty_precision = pi->tick_qty().max_precision();
      int64_t price_multiplier = std::pow(10, price_precision);
      int64_t qty_multiplier = std::pow(10, qty_precision);
      precision_multiplier_cache_.emplace(ch, PrecisionMultiplier{
          price_precision,
          qty_precision,
          price_multiplier,
          qty_multiplier
      });
    } catch (...) {
      int precision = Digits10(default_multiplier_) - 1;
      precision_multiplier_cache_.emplace(ch, PrecisionMultiplier{
          precision,
          precision,
          default_multiplier_,
          default_multiplier_
      });
    }
    return precision_multiplier_cache_[ch];
  }
}

void FastFeedSubConverter::CheckAndUpdatePricePrecision(
    const Channel& ch,
    std::optional<bool>& price_ascending,
    int64_t price_multiplier_new) {
  auto iter = precision_multiplier_cache_.find(ch);
  if (iter != precision_multiplier_cache_.end()) {
    bool ascending = price_multiplier_new > iter->second.price_multiplier;
    if (price_ascending && !price_ascending.value()) {
      CHECK_EQ(price_ascending.value(), ascending) << "multiplier can't back";
    } else {
      price_ascending = ascending;
    }
    iter->second.price_multiplier = price_multiplier_new;
  }
}

void FastFeedSubConverter::CheckAndUpdateQtyPrecision(
    const Channel& ch,
    std::optional<bool>& qty_ascending,
    int64_t qty_multiplier_new) {
  auto iter = precision_multiplier_cache_.find(ch);
  if (iter != precision_multiplier_cache_.end()) {
    bool ascending = qty_multiplier_new > iter->second.qty_multiplier;
    if (qty_ascending && !qty_ascending.value()) {
      CHECK_EQ(qty_ascending.value(), ascending) << "multiplier can't back";
    } else {
      qty_ascending = ascending;
    }
    iter->second.qty_multiplier = qty_multiplier_new;
  }
}

void FastFeedSubConverter::onTradeFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }

  std::optional<int64_t> price_multiplier_new;
  std::optional<int64_t> qty_multiplier_new;
  ConvertTradeProto(upd, price_multiplier_new, qty_multiplier_new);
  if (price_multiplier_new || qty_multiplier_new) {
    if (price_multiplier_new.value_or(1) <= 0 || qty_multiplier_new.value_or(1) <= 0) {
      LOG_EVERY_N(ERROR, 100) << "Invalid trade at " << upd.timestamp();
      return;  // skip invalid value
    }
    WritePrecisionMultiplier(upd, price_multiplier_new, qty_multiplier_new);

    // re convert again
    price_multiplier_new.reset();
    qty_multiplier_new.reset();
    ConvertTradeProto(upd, price_multiplier_new, qty_multiplier_new);
    CHECK(!(price_multiplier_new || qty_multiplier_new));
  }

  writer_->Write(&proto_);
}

void FastFeedSubConverter::ConvertTradeProto(
    const FeedUpdate& upd,
    std::optional<int64_t>& price_multiplier_new,
    std::optional<int64_t>& qty_multiplier_new) {
  auto trade = upd.trade();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  auto& builder = GetFastFeedBuilder(ch, writer, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADE);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADE);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trade.timestamp);

  auto ptrade = proto.mutable_trade();
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);
  int64_t price_multiplier = precision_multiplier.price_multiplier;
  int64_t qty_multiplier = precision_multiplier.qty_multiplier;
  int64_t price_multiplier_guess = std::numeric_limits<int64_t>::quiet_NaN();
  int64_t qty_multiplier_guess = std::numeric_limits<int64_t>::quiet_NaN();
  int64_t price, qty;
  if (ToIntWithGuessPrecions(ch, builder.precision_price_ascending,
                             trade.price, price_multiplier,
                             price, price_multiplier_guess)) {
    CheckAndUpdatePricePrecision(ch,
                                 builder.precision_price_ascending,
                                 price_multiplier_guess);
    price_multiplier = price_multiplier_guess;
    price_multiplier_new = price_multiplier_guess;
  }
  if (ToIntWithGuessPrecions(ch, builder.precision_qty_ascending,
                             trade.fill_qty, qty_multiplier,
                             qty, qty_multiplier_guess)) {
    CheckAndUpdateQtyPrecision(ch,
                               builder.precision_qty_ascending,
                               qty_multiplier_guess);
    qty_multiplier = qty_multiplier_guess;
    qty_multiplier_new = qty_multiplier_guess;
  }
  ptrade->set_price(price);
  ptrade->set_qty(qty);

  if (trade.side == coin2::exchange::feed::TRADE_BUY_SIDE) {
    ptrade->set_side(fastfeed::proto::TRADE_SIDE_BUY);
  } else if (trade.side == coin2::exchange::feed::TRADE_SELL_SIDE) {
    ptrade->set_side(fastfeed::proto::TRADE_SIDE_SELL);
  } else {
    ptrade->set_side(fastfeed::proto::TRADE_SIDE_UNKNOWN);
  }
}

void FastFeedSubConverter::ConvertBookProto(
    const FeedUpdate& upd,
    std::optional<int64_t>& price_multiplier_new,
    std::optional<int64_t>& qty_multiplier_new) {
  auto& book = upd.book();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_BOOK);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(book.Timestamp());

  auto ch = Channel {upd.mea(), symbol};
  auto& builder = GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);
  int64_t price_multiplier = precision_multiplier.price_multiplier;
  int64_t qty_multiplier = precision_multiplier.qty_multiplier;
  int64_t price_multiplier_guess = std::numeric_limits<int64_t>::quiet_NaN();
  int64_t qty_multiplier_guess = std::numeric_limits<int64_t>::quiet_NaN();
  if (publish_bbo_book_only_ ||
      publish_bbo_price_change_ ||
      book.IsSnapshot() ||
      upd.timestamp() - builder.last_snapshot_ts > SNAPSHOT_PERIOD) {
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_SNAPSHOT);
    // snapshot
    auto pbook = proto.mutable_book_snapshot();
    int book_level = 0;
    for (auto& pair : book.Asks()) {
      if (publish_bbo_book_only_) {
        if (book_level >= publish_book_level_) break;
      }

      int64_t price, qty;
      if (ToIntWithGuessPrecions(ch, builder.precision_price_ascending,
                                 pair.first, price_multiplier,
                                 price, price_multiplier_guess)) {
        if (price_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdatePricePrecision(ch,
                                     builder.precision_price_ascending,
                                     price_multiplier_guess);
        price_multiplier = price_multiplier_guess;
        price_multiplier_new = price_multiplier_guess;
      }
      if (ToIntWithGuessPrecions(ch, builder.precision_qty_ascending,
                                 pair.second, qty_multiplier,
                                 qty, qty_multiplier_guess)) {
        if (qty_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdateQtyPrecision(ch,
                                   builder.precision_qty_ascending,
                                   qty_multiplier_guess);
        qty_multiplier = qty_multiplier_guess;
        qty_multiplier_new = qty_multiplier_guess;
      }
      auto level = pbook->add_ask();
      level->set_price(price);
      if (publish_bbo_price_change_ && book_level == 0) {
        last_bbo_prices_[symbol].second = pair.first;
      }
      level->set_qty(qty);
      book_level++;
    }
    book_level = 0;
    for (auto& pair : book.Bids()) {
      if (publish_bbo_book_only_) {
        if (book_level >= publish_book_level_) break;
      }
      int64_t price, qty;
      if (ToIntWithGuessPrecions(ch, builder.precision_price_ascending,
                                 pair.first, price_multiplier,
                                 price, price_multiplier_guess)) {
        if (price_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdatePricePrecision(ch,
                                     builder.precision_price_ascending,
                                     price_multiplier_guess);
        price_multiplier = price_multiplier_guess;
        price_multiplier_new = price_multiplier_guess;
      }
      if (ToIntWithGuessPrecions(ch, builder.precision_qty_ascending,
                                 pair.second, qty_multiplier,
                                 qty, qty_multiplier_guess)) {
        if (qty_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdateQtyPrecision(ch,
                                   builder.precision_qty_ascending,
                                   qty_multiplier_guess);
        qty_multiplier = qty_multiplier_guess;
        qty_multiplier_new = qty_multiplier_guess;
      }
      auto level = pbook->add_bid();
      level->set_price(price);
      if (publish_bbo_price_change_ && book_level == 0) {
        last_bbo_prices_[symbol].first = pair.first;
      }
      level->set_qty(qty);
      book_level++;
    }
  } else {
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_UPDATE);
    // diff
    auto pbook = proto.mutable_book_update();
    for (auto pq : book.DeltaAsks()) {
      auto level = pbook->add_ask();
      int64_t price, qty;
      if (ToIntWithGuessPrecions(ch, builder.precision_price_ascending,
                                 pq.price, price_multiplier,
                                 price, price_multiplier_guess)) {
        if (price_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdatePricePrecision(ch,
                                     builder.precision_price_ascending,
                                     price_multiplier_guess);
        price_multiplier = price_multiplier_guess;
        price_multiplier_new = price_multiplier_guess;
      }
      if (ToIntWithGuessPrecions(ch, builder.precision_qty_ascending,
                                 pq.qty, qty_multiplier,
                                 qty, qty_multiplier_guess)) {
        if (qty_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdateQtyPrecision(ch,
                                   builder.precision_qty_ascending,
                                   qty_multiplier_guess);
        qty_multiplier = qty_multiplier_guess;
        qty_multiplier_new = qty_multiplier_guess;
      }
      level->set_price(price);
      level->set_qty(qty);
    }
    for (auto pq : book.DeltaBids()) {
      auto level = pbook->add_bid();
      int64_t price, qty;
      if (ToIntWithGuessPrecions(ch, builder.precision_price_ascending,
                                 pq.price, price_multiplier,
                                 price, price_multiplier_guess)) {
        if (price_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdatePricePrecision(ch,
                                     builder.precision_price_ascending,
                                     price_multiplier_guess);
        price_multiplier = price_multiplier_guess;
        price_multiplier_new = price_multiplier_guess;
      }
      if (ToIntWithGuessPrecions(ch, builder.precision_qty_ascending,
                                 pq.qty, qty_multiplier,
                                 qty, qty_multiplier_guess)) {
        if (qty_multiplier_guess <= 0) continue;  // skip invalid value
        CheckAndUpdateQtyPrecision(ch,
                                   builder.precision_qty_ascending,
                                   qty_multiplier_guess);
        qty_multiplier = qty_multiplier_guess;
        qty_multiplier_new = qty_multiplier_guess;
      }
      level->set_price(price);
      level->set_qty(qty);
    }
  }
}

void FastFeedSubConverter::onBookFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  if (publish_bbo_price_change_) {
    auto last_bbo_price = last_bbo_prices_[upd.product().absolute_norm()];
    if (!IsBboPriceChanged(upd, last_bbo_price)) {
      return;
    }
  }

  // book contains many price/qty items, have to try several times if multiplier changed
  // at most try 3 times
  std::optional<int64_t> price_multiplier_last_new;
  std::optional<int64_t> qty_multiplier_last_new;
  std::optional<int64_t> price_multiplier_new;
  std::optional<int64_t> qty_multiplier_new;
  int try_times = 0;
  do {
    try_times++;
    price_multiplier_new.reset();
    qty_multiplier_new.reset();
    ConvertBookProto(upd, price_multiplier_new, qty_multiplier_new);
    if (price_multiplier_new) {
      price_multiplier_last_new = price_multiplier_new;
    }
    if (qty_multiplier_new) {
      qty_multiplier_last_new = qty_multiplier_new;
    }
  } while (price_multiplier_new || qty_multiplier_new);
  CHECK_LE(try_times, 3);
  if (price_multiplier_last_new || qty_multiplier_last_new) {
    WritePrecisionMultiplier(upd, price_multiplier_last_new, qty_multiplier_last_new);
  }

  auto symbol = upd.product().absolute_norm();
  auto ch = Channel {upd.mea(), symbol};
  auto& builder = GetFastFeedBuilder(ch, writer_.get(), upd);
  builder.last_snapshot_ts = upd.timestamp();

  writer_->Write(&proto_);
}

void FastFeedSubConverter::WritePrecisionMultiplier(
    const FeedUpdate& upd,
    const std::optional<int64_t>& price_multiplier_new,
    const std::optional<int64_t>& qty_multiplier_new)  {

  auto symbol = upd.product().absolute_norm();
  auto ch = Channel {upd.mea(), symbol};
  auto iter = precision_multiplier_cache_.find(ch);
  CHECK(iter != precision_multiplier_cache_.end());
  std::stringstream info;
  info << upd.mea().String() << "." << symbol << " multipler update, ";
  if (price_multiplier_new) {
    info << "price multiplier " << price_multiplier_new.value();
    iter->second.price_multiplier = price_multiplier_new.value();
  }
  iter->second.price_precision = Digits10(iter->second.price_multiplier) - 1;

  if (qty_multiplier_new) {
    info << "qty multiplier " << qty_multiplier_new.value();
    iter->second.qty_multiplier = qty_multiplier_new.value();
  }
  iter->second.qty_precision = Digits10(iter->second.qty_multiplier) - 1;

  LOG(INFO) << info.str();

  onNewProduct(
        writer_.get(), symbol, upd.timestamp(),
        iter->second.price_precision, iter->second.qty_precision,
        default_multiplier_);
}

void FastFeedSubConverter::onLiquidationFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto liquidation = upd.liquidation_order();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_LIQUIDATION_ORDER);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_LIQUIDATION_ORDER);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(liquidation.timestamp);

  auto pliquidation = proto.mutable_liquidation();
  pliquidation->set_price(ToInt(liquidation.price, precision_multiplier.price_multiplier));
  pliquidation->set_qty(ToInt(liquidation.qty, precision_multiplier.qty_multiplier));
  pliquidation->set_order_id(liquidation.order_id);

  if (liquidation.order_side == coin::proto::BUY_ORDER ||
      liquidation.order_side == coin::proto::BUY_CLOSE_ORDER ) {
    pliquidation->set_direction(fastfeed::proto::ORDER_DIRECTION_BUY);
  } else if (liquidation.order_side == coin::proto::SELL_ORDER ||
              liquidation.order_side == coin::proto::SELL_CLOSE_ORDER) {
    pliquidation->set_direction(fastfeed::proto::ORDER_DIRECTION_SELL);
  } else if (liquidation.order_side == coin::proto::UNKNOWN_ORDER_SIDE) {
    pliquidation->set_direction(fastfeed::proto::ORDER_DIRECTION_UNKNOWN);
  } else {
    THROW() << "NOT support liquidation order side " << liquidation.order_side;
  }

  writer->Write(&proto);
}

void FastFeedSubConverter::onOpenInterestFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& open_interest = upd.open_interest();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_OPEN_INTEREST);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_OPEN_INTEREST);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(open_interest.timestamp);

  auto popen_interest = proto.mutable_open_interest();
  popen_interest->set_qty(ToInt(
      open_interest.open_interest_qty, precision_multiplier.qty_multiplier));

  writer->Write(&proto);
}

void FastFeedSubConverter::onIndexFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& index = upd.index();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_INDEX);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_INDEX);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(index.timestamp);

  auto pindex = proto.mutable_index();
  pindex->set_price(ToInt(index.price, precision_multiplier.price_multiplier));

  writer->Write(&proto);
}

void FastFeedSubConverter::onMarkPriceFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& mark_price = upd.mark_price();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_MARK_PRICE);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_MARK_PRICE);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(mark_price.timestamp);

  auto pmark_price = proto.mutable_mark_price();
  pmark_price->set_price(ToInt(mark_price.mark_price, precision_multiplier.price_multiplier));

  writer->Write(&proto);
}

void FastFeedSubConverter::onFundingRateFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& funding_rate = upd.funding_rate();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  {
    auto ch = Channel {upd.mea(), symbol};
    GetFastFeedBuilder(ch, writer, upd);
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_FUNDING_RATE);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_FUNDING_RATE);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(funding_rate.timestamp);

  auto pfunding_rate = proto.mutable_funding_rate();
  pfunding_rate->set_funding_rate(ToInt(funding_rate.funding_rate, default_multiplier_));
  pfunding_rate->set_funding_time(funding_rate.funding_time);
  pfunding_rate->set_estimated_rate(ToInt(funding_rate.estimated_rate, default_multiplier_));

  writer->Write(&proto);
}

void FastFeedSubConverter::onIvGreeksFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& iv_greeks = upd.iv_greeks();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  {
    auto ch = Channel {upd.mea(), symbol};
    GetFastFeedBuilder(ch, writer, upd);
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_IV_GREEKS);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_IV_GREEKS);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(iv_greeks.timestamp);

  auto piv_greeks = proto.mutable_iv_greeks();
  piv_greeks->set_ask_iv(ToInt(iv_greeks.ask_iv, default_multiplier_));
  piv_greeks->set_bid_iv(ToInt(iv_greeks.bid_iv, default_multiplier_));
  piv_greeks->set_mark_iv(ToInt(iv_greeks.mark_iv, default_multiplier_));

  piv_greeks->set_delta(ToInt(iv_greeks.delta, default_multiplier_));
  piv_greeks->set_gamma(ToInt(iv_greeks.gamma, default_multiplier_));
  piv_greeks->set_rho(ToInt(iv_greeks.rho, default_multiplier_));
  piv_greeks->set_theta(ToInt(iv_greeks.theta, default_multiplier_));
  piv_greeks->set_vega(ToInt(iv_greeks.vega, default_multiplier_));

  writer->Write(&proto);
}

void FastFeedSubConverter::onNavFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto& nav = upd.nav();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);
  const auto& precision_multiplier = GetPrecisionMultiplier(ch, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_NAV);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_NAV);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(nav.timestamp);

  auto pnav = proto.mutable_nav();
  pnav->set_actual_leverage(ToInt(nav.actual_leverage, default_multiplier_));
  pnav->set_nav(ToInt(nav.nav, precision_multiplier.price_multiplier));
  pnav->set_outstanding(ToInt(nav.outstanding, default_multiplier_));
  for (const auto& ua : nav.basket) {
    auto basket = pnav->add_basket();
    basket->set_symbol_or_currency(ua.symbol_or_currency);
    basket->set_amount(ToInt(ua.amount, default_multiplier_));
  }

  writer->Write(&proto);
}

void FastFeedSubConverter::onKlineFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }
  auto klines = upd.klines();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  auto& builder = GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::KlineInterval interval = klines.Get().value()->interval;
  int64_t last_open_timestamp = klines_last_open_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((klines.Type() == Klines::SNAPSHOT) ||
      (upd.timestamp() - builder.last_snapshot_ts > SNAPSHOT_PERIOD)) {
    last_open_timestamp = 0;
    builder.last_snapshot_ts = upd.timestamp();
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_KLINE);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_KLINE);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(klines.Get().value()->timestamp);

  bool new_kline_come_in = false;
  ::fastfeed::proto::Kline* pkline = nullptr;
  for (auto iter = klines.RBegin(); iter != klines.REnd(); iter++) {
    if (is_snapshot) {
      pkline = proto.mutable_kline_snapshot()->add_kline();
    } else {
      // only save new kline to fastfeed
      if (iter->open_timestamp <= last_open_timestamp) continue;

      pkline = proto.mutable_kline_update()->add_kline();
    }
    pkline->set_interval(iter->interval);
    pkline->set_open_timestamp(iter->open_timestamp);
    pkline->set_close_timestamp(iter->close_timestamp);
    pkline->set_open(iter->open);
    pkline->set_close(iter->close);
    pkline->set_high(iter->high);
    pkline->set_low(iter->low);
    if (!std::isnan(iter->volume)) pkline->set_volume(iter->volume);
    if (!std::isnan(iter->turnover)) pkline->set_turnover(iter->turnover);
    if (!std::isnan(iter->buy_volume)) pkline->set_buy_volume(iter->buy_volume);
    if (!std::isnan(iter->buy_turnover)) pkline->set_buy_turnover(iter->buy_turnover);
    klines_last_open_timestamp_[{symbol, interval}] = iter->open_timestamp;
    new_kline_come_in = true;
  }

  if (new_kline_come_in) {
    CHECK(nullptr != pkline) << "No klines data";
    writer->Write(&proto);
  }
}

void FastFeedSubConverter::onWalletFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }

  auto& wallet = upd.wallet();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_WALLET);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_WALLET);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(wallet.timestamp);
  
  auto pwallet = proto.mutable_wallet();
  pwallet->set_can_deposit(wallet.can_deposit);
  pwallet->set_can_withdraw(wallet.can_withdraw);
  pwallet->set_blockchain_delay(wallet.blockchain_delay);

  writer->Write(&proto);
}

void FastFeedSubConverter::onMarketAnalysisFeed(const FeedUpdate& upd)  {
  if (upd.timestamp() < start_from_) {
    return;
  }

  auto& market_analysis = upd.market_analysis();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_MARKET);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_MARKET);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(market_analysis.timestamp);
  
  auto pmarket_analysis = proto.mutable_market_analysis();
  pmarket_analysis->set_total_supply(market_analysis.total_supply);
  pmarket_analysis->set_max_supply(market_analysis.max_supply);
  pmarket_analysis->set_circulating_supply(market_analysis.circulating_supply);
  pmarket_analysis->set_market_cap_in_usd(market_analysis.market_cap_in_usd);
  pmarket_analysis->set_tvl_in_usd(market_analysis.tvl_in_usd);
  pmarket_analysis->set_total_volume_in_usd(market_analysis.total_volume_in_usd);
  ::fastfeed::proto::VolumePerExchange* pvolume_per_exchange = nullptr;
  for (auto it = market_analysis.volume_per_exchange_list.begin(); it != market_analysis.volume_per_exchange_list.end(); it++) {
    pvolume_per_exchange = pmarket_analysis->add_volume_per_exchange_list();
    pvolume_per_exchange->set_exchange_name(it->exchange_name);
    pvolume_per_exchange->set_volume(it->volume);
    pvolume_per_exchange->set_volume_in_usd(it->volume_in_usd);
  }
  writer->Write(&proto);
}

void FastFeedSubConverter::onTopLongShortAccountRatioFeed(const FeedUpdate& upd) {
  if (upd.timestamp() < start_from_) {
    return;
  }
  const auto& trading_data = upd.top_long_short_account_ratio();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::TradingDataInterval interval = trading_data.Interval();
  int64_t last_timestamp = top_account_last_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((trading_data.Type() == TradingData<TopLongShortAccountRatio>::SNAPSHOT) ||
      (upd.timestamp() - last_timestamp > SNAPSHOT_PERIOD)) {
    last_timestamp = 0;
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trading_data.Timestamp());

  bool new_come_in = false;
  ::fastfeed::proto::LongShortRatio* pratio = nullptr;
  for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
    if (is_snapshot) {
      auto* snapshot = proto.mutable_trading_data_snapshot();
      snapshot->set_interval(interval);
      pratio = snapshot->add_top_long_short_account_ratio();
    } else {
      // only save new kline to fastfeed
      if (iter->timestamp <= last_timestamp) continue;

      auto* update = proto.mutable_trading_data_update();
      update->set_interval(interval);
      pratio = update->add_top_long_short_account_ratio();
    }
    pratio->set_timestamp(iter->timestamp);
    pratio->set_long_short_ratio(iter->long_short_ratio);
    pratio->set_long_(iter->long_account);
    pratio->set_short_(iter->short_account);

    top_account_last_timestamp_[{symbol, interval}] = iter->timestamp;
    new_come_in = true;
  }

  if (new_come_in) {
    CHECK(nullptr != pratio) << "No top long short account ratio data";
    writer->Write(&proto);
  }
}

void FastFeedSubConverter::onTopLongShortPositionRatioFeed(const FeedUpdate& upd) {
  if (upd.timestamp() < start_from_) {
    return;
  }
  const auto& trading_data = upd.top_long_short_position_ratio();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::TradingDataInterval interval = trading_data.Interval();
  int64_t last_timestamp = top_position_last_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((trading_data.Type() == TradingData<TopLongShortPositionRatio>::SNAPSHOT) ||
      (upd.timestamp() - last_timestamp > SNAPSHOT_PERIOD)) {
    last_timestamp = 0;
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trading_data.Timestamp());

  bool new_come_in = false;
  ::fastfeed::proto::LongShortRatio* pratio = nullptr;
  for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
    if (is_snapshot) {
      auto* snapshot = proto.mutable_trading_data_snapshot();
      snapshot->set_interval(interval);
      pratio = snapshot->add_top_long_short_position_ratio();
    } else {
      // only save new kline to fastfeed
      if (iter->timestamp <= last_timestamp) continue;

      auto* update = proto.mutable_trading_data_update();
      update->set_interval(interval);
      pratio = update->add_top_long_short_position_ratio();
    }
    pratio->set_timestamp(iter->timestamp);
    pratio->set_long_short_ratio(iter->long_short_ratio);
    pratio->set_long_(iter->long_position);
    pratio->set_short_(iter->short_position);

    top_position_last_timestamp_[{symbol, interval}] = iter->timestamp;
    new_come_in = true;
  }

  if (new_come_in) {
    CHECK(nullptr != pratio) << "No top long short account ratio data";
    writer->Write(&proto);
  }
}

void FastFeedSubConverter::onGlobalLongShortAccountRatioFeed(const FeedUpdate& upd) {
  if (upd.timestamp() < start_from_) {
    return;
  }
  const auto& trading_data = upd.global_long_short_account_ratio();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::TradingDataInterval interval = trading_data.Interval();
  int64_t last_timestamp = global_account_last_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((trading_data.Type() == TradingData<GlobalLongShortAccountRatio>::SNAPSHOT) ||
      (upd.timestamp() - last_timestamp > SNAPSHOT_PERIOD)) {
    last_timestamp = 0;
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trading_data.Timestamp());

  bool new_come_in = false;
  ::fastfeed::proto::LongShortRatio* pratio = nullptr;
  for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
    if (is_snapshot) {
      auto* snapshot = proto.mutable_trading_data_snapshot();
      snapshot->set_interval(interval);
      pratio = snapshot->add_global_long_short_account_ratio();
    } else {
      // only save new kline to fastfeed
      if (iter->timestamp <= last_timestamp) continue;

      auto* update = proto.mutable_trading_data_update();
      update->set_interval(interval);
      pratio = update->add_global_long_short_account_ratio();
    }
    pratio->set_timestamp(iter->timestamp);
    pratio->set_long_short_ratio(iter->long_short_ratio);
    pratio->set_long_(iter->long_account);
    pratio->set_short_(iter->short_account);

    global_account_last_timestamp_[{symbol, interval}] = iter->timestamp;
    new_come_in = true;
  }

  if (new_come_in) {
    CHECK(nullptr != pratio) << "No top long short account ratio data";
    writer->Write(&proto);
  }
}

void FastFeedSubConverter::onTakerBuySellRatioFeed(const FeedUpdate& upd) {
  if (upd.timestamp() < start_from_) {
    return;
  }
  const auto& trading_data = upd.taker_buy_sell_ratio();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::TradingDataInterval interval = trading_data.Interval();
  int64_t last_timestamp = taker_volume_last_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((trading_data.Type() == TradingData<TakerBuySellRatio>::SNAPSHOT) ||
      (upd.timestamp() - last_timestamp > SNAPSHOT_PERIOD)) {
    last_timestamp = 0;
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trading_data.Timestamp());

  bool new_come_in = false;
  ::fastfeed::proto::TakerBuySellRatio* pratio = nullptr;
  for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
    if (is_snapshot) {
      auto* snapshot = proto.mutable_trading_data_snapshot();
      snapshot->set_interval(interval);
      pratio = snapshot->add_taker_buy_sell_ratio();
    } else {
      // only save new kline to fastfeed
      if (iter->timestamp <= last_timestamp) continue;

      auto* update = proto.mutable_trading_data_update();
      update->set_interval(interval);
      pratio = update->add_taker_buy_sell_ratio();
    }
    pratio->set_timestamp(iter->timestamp);
    pratio->set_buy_sell_ratio(iter->buy_sell_ratio);
    pratio->set_buy_volume(iter->buy_volume);
    pratio->set_sell_volume(iter->sell_volume);

    taker_volume_last_timestamp_[{symbol, interval}] = iter->timestamp;
    new_come_in = true;
  }

  if (new_come_in) {
    CHECK(nullptr != pratio) << "No top long short account ratio data";
    writer->Write(&proto);
  }
}

void FastFeedSubConverter::onOpenInterestHistFeed(const FeedUpdate& upd) {
  if (upd.timestamp() < start_from_) {
    return;
  }
  const auto& trading_data = upd.open_interest_hist();
  auto writer = writer_.get();
  auto symbol = upd.product().absolute_norm();

  // if first event, write metadata first
  auto ch = Channel {upd.mea(), symbol};
  GetFastFeedBuilder(ch, writer, upd);

  fastfeed::proto::TradingDataInterval interval = trading_data.Interval();
  int64_t last_timestamp = open_interest_hist_last_timestamp_[{symbol, interval}];
  bool is_snapshot = false;
  if ((trading_data.Type() == TradingData<OpenInterest>::SNAPSHOT) ||
      (upd.timestamp() - last_timestamp > SNAPSHOT_PERIOD)) {
    last_timestamp = 0;
    is_snapshot = true;
  }

  auto& proto = proto_;
  proto.Clear();
  proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
  proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
  proto.set_symbol(symbol);
  proto.set_timestamp(upd.timestamp());
  proto.set_origin_timestamp(trading_data.Timestamp());

  bool new_come_in = false;
  ::fastfeed::proto::OpenInterest* popen_interest = nullptr;
  for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
    if (is_snapshot) {
      auto* snapshot = proto.mutable_trading_data_snapshot();
      snapshot->set_interval(interval);
      popen_interest = snapshot->add_open_interest_hist();
    } else {
      // only save new kline to fastfeed
      if (iter->timestamp <= last_timestamp) continue;

      auto* update = proto.mutable_trading_data_update();
      update->set_interval(interval);
      popen_interest = update->add_open_interest_hist();
    }
    popen_interest->set_timestamp(iter->timestamp);
    popen_interest->set_qty_float(iter->open_interest_qty);

    open_interest_hist_last_timestamp_[{symbol, interval}] = iter->timestamp;
    new_come_in = true;
  }

  if (new_come_in) {
    CHECK(nullptr != popen_interest) << "No top long short account ratio data";
    writer->Write(&proto);
  }
}


FastFeedConverter::FastFeedConverter(
    const std::string& out_dir,
    int multiplier,
    const std::string& pi_dir)
    : out_dir_(out_dir),
      pi_dir_(pi_dir),
      default_multiplier_(multiplier) {
}

void FastFeedConverter::Init(::coin2::app::Driver* driver) {
  this->Init(&(driver->config()));
}

void FastFeedConverter::Init(const ::coin2::app::DriverConfig* driver_config) {
  // Converter only support single mea execution.
  const auto& config = driver_config->feed();
  auto it = config.exchanges().begin();
  auto& exchange_config = it->second;
  for (const auto& recipe : exchange_config.archive_recipes()) {
    auto sub_converter = std::make_unique<FastFeedSubConverter>(
        recipe, out_dir_, default_multiplier_ , pi_dir_);
    sub_converter->Init(driver_config);
    sub_converters_[recipe] = std::move(sub_converter);
  }
}

void FastFeedConverter::onTradeFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onTradeFeed(upd);
}

void FastFeedConverter::onBookFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onBookFeed(upd);
}

void FastFeedConverter::onLiquidationFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onLiquidationFeed(upd);
}

void FastFeedConverter::onOpenInterestFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onOpenInterestFeed(upd);
}

void FastFeedConverter::onIndexFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onIndexFeed(upd);
}

void FastFeedConverter::onMarkPriceFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onMarkPriceFeed(upd);
}

void FastFeedConverter::onFundingRateFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onFundingRateFeed(upd);
}

void FastFeedConverter::onIvGreeksFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onIvGreeksFeed(upd);
}

void FastFeedConverter::onNavFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onNavFeed(upd);
}

void FastFeedConverter::onKlineFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onKlineFeed(upd);
}

void FastFeedConverter::onWalletFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onWalletFeed(upd);
}

void FastFeedConverter::onMarketAnalysisFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onMarketAnalysisFeed(upd);
}

void FastFeedConverter::onTopLongShortAccountRatioFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onTopLongShortAccountRatioFeed(upd);
}

void FastFeedConverter::onTopLongShortPositionRatioFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onTopLongShortPositionRatioFeed(upd);
}

void FastFeedConverter::onGlobalLongShortAccountRatioFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onGlobalLongShortAccountRatioFeed(upd);
}

void FastFeedConverter::onTakerBuySellRatioFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onTakerBuySellRatioFeed(upd);
}

void FastFeedConverter::onOpenInterestHistFeed(const FeedUpdate& upd)  {
  const auto& recipe = upd.recipe();
  sub_converters_[recipe]->onOpenInterestHistFeed(upd);
}
